package beautiful.butterfly.distributed_task_processing_engine.task_manage.task_processor;

import beautiful.butterfly.distributed_task_processing_engine.api.TaskDataItemCollectorAndExecuter;
import beautiful.butterfly.distributed_task_processing_engine.cron_expression.CronExpression;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.JvmInstanceStatusAndTaskManage;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.ForEverRunTaskDataItemProcessor;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.ITaskDataItemProcessor;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.WaitAllFinishTaskDataItemProcessor;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.task_data_item_data_manager.ITaskAndTaskProcessorAndTaskDataItemManage;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.task_data_item_data_manager.TaskItemStringAndParameterString;
import beautiful.butterfly.distributed_task_processing_engine.tools.DateFormats;
import beautiful.butterfly.distributed_task_processing_engine.tools.Name;
import beautiful.butterfly.distributed_task_processing_engine.user_definition.TaskUserDefinition;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.data.Stat;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 1、任务调度分配器的目标：	让所有的任务不重复，不遗漏的被快速处理。
 * 2、一个Manager只管理一种任务类型的一组工作线程。
 * 3、在一个JVM里面可能存在多个处理相同任务类型的Manager，也可能存在处理不同任务类型的Manager。
 * 4、在不同的JVM里面可以存在处理相同任务的Manager
 * 5、调度的Manager可以动态的随意增加和停止
 * <p/>
 * 主要的职责：
 * 1、定时向集中的数据配置中心更新当前调度服务器的心跳状态
 * 2、向数据配置中心获取所有服务器的状态来重新计算任务的分配。这么做的目标是避免集中任务调度中心的单点问题。
 * 3、在每个批次数据处理完毕后，检查是否有其它处理服务器申请自己把持的任务队列，如果有，则释放给相关处理服务器。
 * <p/>
 * 其它：
 * 如果当前服务器在处理当前任务的时候超时，需要清除当前队列，并释放已经把持的任务。并向控制主动中心报警。
 */

@Slf4j
@Name("TaskProcessor")
@SuppressWarnings({"rawtypes", "unchecked"})
public class TaskProcessor implements ITaskProcessor
{

    @Setter
    @Getter
    JvmInstanceStatusAndTaskManage jvmInstanceStatusAndTaskManage;
    //
    @Setter
    @Getter
    public ITaskAndTaskProcessorAndTaskDataItemManage taskAndTaskProcessorAndTaskDataItemManage;

    //任务信息
    @Setter
    @Getter
    protected TaskUserDefinition taskUserDefinition;
    @Setter
    @Getter
    protected TaskProcessorInfo taskProcessorInfo;
    @Setter
    @Getter
    protected Lock registerOrUnRegisterTaskProcessorLock = new ReentrantLock();
    @Setter
    @Getter
    public ITaskDataItemProcessor taskDataItemProcessor;
    @Setter
    @Getter
    TaskDataItemCollectorAndExecuter taskDataItemCollectorAndExecuter;
    @Setter
    @Getter
    protected boolean minTaskProcessorIdIsControlTaskProcessorId = false;//是否稳定
    @Setter
    @Getter
    private Timer heartBeatCheckAndReallocateTaskTimer;
    //默认是停止调度标志,只能先开始执行任务变成非停止调度标志,才能停止调度.如果状态随意改变,则不能控制停止和恢复状态.且一个任务调度的开始时间必须早于其结束时间.
    @Setter
    @Getter
    boolean isPauseSchedule = true;//暂停调度
    @Setter
    @Getter
    protected boolean isStopSchedule = false;//停止调度
    Statistics statistics = new Statistics();
    //获取任务项
    @Setter
    @Getter
    protected List<TaskItemStringAndParameterString> taskItemStringAndParameterStringList = new CopyOnWriteArrayList<TaskItemStringAndParameterString>();
    @Setter
    @Getter
    protected boolean needGetTaskItemStringAndParameterStringList = true;
    @Setter
    @Getter
    protected long getTaskItemStringAndParameterStringListTimeMillis = 0;
    @Setter
    @Getter
    private final Object getTaskItemStringAndParameterStringListLock = new Object();
    @Setter
    @Getter
    protected int taskItemCount = 0;
    @Setter
    @Getter
    protected long lastFetchVersion = -1;

    /**
     * 当前线程组编号
     */
    public static final AtomicInteger serialNumber = new AtomicInteger(1);
    @Setter
    @Getter
    protected int currentThreadNo = 0;

    public TaskProcessor(JvmInstanceStatusAndTaskManage jvmInstanceStatusAndTaskManage, String scheduleTaskId, ITaskAndTaskProcessorAndTaskDataItemManage taskAndTaskProcessorAndTaskDataItemManage) throws Exception
    {
        this.jvmInstanceStatusAndTaskManage = jvmInstanceStatusAndTaskManage;
        this.currentThreadNo = serialNumber.incrementAndGet();
        this.taskAndTaskProcessorAndTaskDataItemManage = taskAndTaskProcessorAndTaskDataItemManage;
        this.taskUserDefinition = this.taskAndTaskProcessorAndTaskDataItemManage.getTaskUserDefinition(scheduleTaskId);

        this.taskAndTaskProcessorAndTaskDataItemManage.clearExpireTask(scheduleTaskId, this.taskUserDefinition.getExpireOwnSignInterval());
        //check bean
        Object bean = jvmInstanceStatusAndTaskManage.getBean(this.taskUserDefinition.getBeanName());
        if (bean == null)
        {
            throw new Exception("this.taskUserDefinition.getBeanName() " + "不存在");
        }
        if (!(bean instanceof TaskDataItemCollectorAndExecuter))
        {
            throw new Exception("bean" + this.taskUserDefinition.getBeanName() + "不是" + TaskDataItemCollectorAndExecuter.class.getSimpleName() + "的实例");
        }
        this.taskDataItemCollectorAndExecuter = (TaskDataItemCollectorAndExecuter) bean;
        //
        if (this.taskUserDefinition.getJudgeDeadInterval() < this.taskUserDefinition.getHeartBeatRate() * 5)
        {
            throw new Exception("JudgeDeadInterval = " + this.taskUserDefinition.getJudgeDeadInterval() + ",HeartBeatRate = " + this.taskUserDefinition.getHeartBeatRate());
        }
        //

        this.taskProcessorInfo = TaskProcessorInfo.createTaskProcessorInfo(this.taskAndTaskProcessorAndTaskDataItemManage, jvmInstanceStatusAndTaskManage.getJvmInstanceId(), scheduleTaskId, this.taskUserDefinition.getThreadNumber());
        this.taskProcessorInfo.setJvmInstanceId(this.jvmInstanceStatusAndTaskManage.getJvmInstanceId());
        //修改了taskProcessorid,该方法内部已经设置了taskProcessorInfo.setRegister(true);
        this.taskAndTaskProcessorAndTaskDataItemManage.registerTaskProcessorInfoWithSerialNumberAndSetRegisterTrue(this.taskProcessorInfo);


        this.heartBeatCheckAndReallocateTaskTimer = new Timer(this.taskProcessorInfo.getTaskId() + "-" + this.currentThreadNo + "-HeartBeat");
        this.heartBeatCheckAndReallocateTaskTimer.schedule(new HeartBeatCheckAndReallocateTaskTimerTask(this),
                new Date(System.currentTimeMillis() + 500),
                this.taskUserDefinition.getHeartBeatRate());
        newInitThread();
    }


    public void init()
    {
        //没有实现的方法，需要的参数直接从任务配置中读取
    }

    /**
     * 当服务器停止的时候，调用此方法清除所有未处理任务，清除服务器的注册信息。
     * 也可能是控制中心发起的终止指令。
     * 需要注意的是，这个方法必须在当前任务处理完毕后才能执行
     * <p>
     * cause:从生产环境实践发现，在zk不稳定且单机job多、调度间隔小的情况下。大量的线程不能正确
     * 地进行状态扭转。官方3.3.3.2版本仍未处理此问题，此处不使用官方的isPauseSchedule字段，其
     * 在发生线程问题并且由于线程问题导致strategy仅剩1个时不能调用stop方法，由此导致该strategy
     * 死循环调用resume方法造成大量日志。
     */
    public void stop() throws Exception
    {
        this.isPauseSchedule = false;
        if (this.taskDataItemProcessor != null)
        {
            this.taskDataItemProcessor.setStopScheduleFlagAndClearTaskList();
        } else
        {
            this.unRegisterTaskProcessor();
        }
    }

    //新建一个初始线程,大概做一些任务分配的事情,后期交给心跳线程进行管理
    public void newInitThread() throws Exception
    {
        new Thread()
        {

            public void run()
            {
                try
                {
                    //不满足条件则需要一直循环,直到任务处理器稳定，这个时候分配出来的任务项才是正确的。
                    while (!minTaskProcessorIdIsControlTaskProcessorId)
                    {
                        if (isStopSchedule)
                        {

                            log.warn("处理器停止运行:" + taskProcessorInfo.getTaskProcessorId());
                            return;
                        }
                        try
                        {
                            clearExpireTaskProcessorAndReleaseTaskItemsCurrentTaskProcessorIdNotInTaskProcessorIdList();
                            minTaskProcessorIdIsControlTaskProcessorId = taskAndTaskProcessorAndTaskDataItemManage.minTaskProcessorIdIsLeaderTaskProcessorId(taskProcessorInfo.getTaskId());


                        } catch (Throwable e)
                        {//忽略初始化的异常
                            log.error(e.getMessage(), e);
                        }
                        if (!minTaskProcessorIdIsControlTaskProcessorId)
                        {
                            Thread.currentThread().sleep(1000);
                        }
                    }//end

                    //保证minTaskProcessorIdIsControlTaskProcessorId才进行后面的操作
                    getTaskItemStringAndParameterStringListTimeMillis = taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis();
                    while (synchronizedGetTaskItemStringAndParameterStringList().size() <= 0)
                    {
                        if (isStopSchedule)
                        {
                            log.warn("处理器停止运行:" + taskProcessorInfo.getTaskProcessorId());
                            return;
                        }
                        Thread.currentThread().sleep(1000);
                    }

                    taskItemCount = taskAndTaskProcessorAndTaskDataItemManage.getTaskItemList(taskProcessorInfo.getTaskId()).size();
                    computeStartAndEndRunDate();
                } catch (Exception e)
                {
                    log.error(e.getMessage(), e);

                }
            }//异步
        }.start();
    }

    /**
     * 先清理过期处理器,再释放任务项当前处理器不在有效处理器列表的处理器(也相当于过期)
     */
    public void clearExpireTaskProcessorAndReleaseTaskItemsCurrentTaskProcessorIdNotInTaskProcessorIdList() throws Exception
    {
        //每个处理器均有清除处理器的责任
        taskAndTaskProcessorAndTaskDataItemManage.clearExpireTaskProcessor(this.taskProcessorInfo.getTaskId(), this.taskUserDefinition.getJudgeDeadInterval());
        //返回该任务所有的处理器
        List<String> taskProcessorIdList = taskAndTaskProcessorAndTaskDataItemManage.getTaskProcessorIdList(this.taskProcessorInfo.getTaskId());
        if (taskAndTaskProcessorAndTaskDataItemManage.isLeaderTaskProcessorId(taskProcessorIdList, this.taskProcessorInfo.getTaskProcessorId()))
        {//是第一次启动，先清楚所有的垃圾数据
            log.warn(this.taskProcessorInfo.getTaskProcessorId() + ":" + taskProcessorIdList.size());
            this.taskAndTaskProcessorAndTaskDataItemManage.releaseOldTaskItemsAndReAllotTaskItemInTaskProcessorIdList(this.taskProcessorInfo.getTaskId(), this.taskProcessorInfo.getTaskProcessorId());
        }
    }

    /**
     * 由于上面在数据执行时有使用到synchronized ，但是心跳线程并没有对应加锁。
     * 所以在此方法上加一下synchronized
     * <p>
     * 延长调度死亡报警时长，增加僵尸节点报警，然而没有调度服务器重建逻辑。留个todo
     * TODO:尝试恢复serverList
     */
    protected synchronized List<TaskItemStringAndParameterString> synchronizedGetTaskItemStringAndParameterStringList() throws Exception
    {
        try
        {
            Map<String, Stat> taskProcessorIdToStatMap = this.taskAndTaskProcessorAndTaskDataItemManage.getTaskProcessorIdToStatMap(this.taskProcessorInfo.getTaskId());
            isExistTimeoutTaskProcessor(this.taskProcessorInfo.getTaskId(), taskProcessorIdToStatMap);
        } catch (Exception e)
        {
            log.error(e.getMessage());
        }
        //获取最新的版本号
        this.lastFetchVersion = this.taskAndTaskProcessorAndTaskDataItemManage.getTaskProcessors$Stat$Version(this.taskProcessorInfo.getTaskId());

        try
        {
            //释放之前占用进行待处理的任务项,在稍后的任务项获取方法中重新获取
            this.taskAndTaskProcessorAndTaskDataItemManage.releaseTaskItemCurrentTaskProcessorId(this.taskProcessorInfo.getTaskId(), this.taskProcessorInfo.getTaskProcessorId());
            //重新查询当前服务器能够处理的队列,为了避免在休眠切换的过程中出现队列瞬间的不一致，先清除内存中的队列
            this.taskItemStringAndParameterStringList.clear();
            this.taskItemStringAndParameterStringList = this.taskAndTaskProcessorAndTaskDataItemManage.getTaskItemStringAndParameterStringListWithTaskProcessorId(this.taskProcessorInfo.getTaskId(), this.taskProcessorInfo.getTaskProcessorId());
            //如果超过10个心跳周期还没有获取到调度队列，则报警
            if (this.taskItemStringAndParameterStringList.size() == 0//
                    &&//
                    taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis() - this.getTaskItemStringAndParameterStringListTimeMillis
                            > this.taskUserDefinition.getHeartBeatRate() * 20)
            {
                StringBuffer sb = new StringBuffer();
                sb.append(this.taskProcessorInfo.getTaskProcessorId());
                sb.append("[TASK_TYPE=");
                sb.append(this.taskProcessorInfo.getTaskId());
                sb.append("]");
                sb.append("  taskItemStringAndParameterStringList.size() =" + taskItemStringAndParameterStringList.size());
                sb.append(" ,taskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()=" + taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis());
                sb.append(" ,getTaskItemStringAndParameterStringListTimeMillis=" + getTaskItemStringAndParameterStringListTimeMillis);
                sb.append(" ,taskId.getHeartBeatRate()=" + taskUserDefinition.getHeartBeatRate() * 10);
                log.warn(sb.toString());
            }
            //更新时间戳
            if (this.taskItemStringAndParameterStringList.size() > 0)
            {
                this.getTaskItemStringAndParameterStringListTimeMillis = taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis();
            }
            return this.taskItemStringAndParameterStringList;//all jvm_instance_info_and_task_processor_allocation_strategy item  id
        } catch (Throwable e)
        {
            this.lastFetchVersion = -1; //必须把把版本号设置小，避免任务加载失败
            if (e instanceof Exception)
            {
                throw (Exception) e;
            } else
            {
                throw new Exception(e);
            }
        }
    }

    /**
     * 判断某个任务对应的线程组是否处于僵尸状态。
     * true 表示有线程组处于僵尸状态。需要告警。
     */
    private boolean isExistTimeoutTaskProcessor(String scheduleTaskId, Map<String, Stat> taskProcessorIdToStatMap) throws Exception
    {
        boolean exist = false;
        for (String taskProcessorId : taskProcessorIdToStatMap.keySet())
        {
            Stat stat = taskProcessorIdToStatMap.get(taskProcessorId);
            if (this.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis() - stat.getMtime() > this.taskUserDefinition.getHeartBeatRate() * 40)
            {
                log.error("zombie serverList exists! serv=" + taskProcessorId + " ,type=" + scheduleTaskId + "");
                exist = true;
                break;
            }
        }
        return exist;

    }

    /**
     * 开始的时候，计算第一次执行时间
     */

    public void computeStartAndEndRunDate() throws Exception
    {
        //只有当存在可执行队列后再开始启动队列
        boolean isRunNow = false;
        if (this.taskUserDefinition.getPlanRunStartTimeString() == null)
        {
            isRunNow = true;
        } else
        {
            String planRunStartTimeString = this.taskUserDefinition.getPlanRunStartTimeString();
            if (planRunStartTimeString.toLowerCase().startsWith("startrun:"))
            {
                isRunNow = true;
                planRunStartTimeString = planRunStartTimeString.substring("startrun:".length());
            }
            CronExpression cronExpressionForPlanRunStartTime = new CronExpression(planRunStartTimeString);
            Date nowDate = new Date(this.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis());
            Date startTimeAfterNowDate = cronExpressionForPlanRunStartTime.getNextTimeAfter(nowDate);
            //
            this.heartBeatCheckAndReallocateTaskTimer.schedule(
                    new PauseOrResumeTimerTask(this, this.heartBeatCheckAndReallocateTaskTimer,
                            PauseOrResumeTimerTask.type_resume, planRunStartTimeString),
                    startTimeAfterNowDate);//在这个时间点开始调用
            this.taskProcessorInfo.setNextRunStartTime(DateFormats.transferDataToString(startTimeAfterNowDate));
            //
            if (this.taskUserDefinition.getPlanRunEndTimeString() == null
                    || this.taskUserDefinition.getPlanRunEndTimeString().equals("-1"))
            {
                this.taskProcessorInfo.setNextRunEndTime("pauseTaskProcessorAllocationStrategy");//运行完毕后终止
            } else
            {
                try
                {
                    String planRunEndTimeString = this.taskUserDefinition.getPlanRunEndTimeString();
                    CronExpression cronExpressionForPlanRunEndTime = new CronExpression(planRunEndTimeString);
                    Date endTimeAfterStartTimeAfterNowDate = cronExpressionForPlanRunEndTime.getNextTimeAfter(startTimeAfterNowDate);
                    Date endTimeAfterNowDate = cronExpressionForPlanRunEndTime.getNextTimeAfter(nowDate);
                    //
                    if (!endTimeAfterNowDate.equals(endTimeAfterStartTimeAfterNowDate)//
                            &&//
                            nowDate.before(endTimeAfterNowDate))//
                    {
                        isRunNow = true;
                        endTimeAfterStartTimeAfterNowDate = endTimeAfterNowDate;//选择最开始的结束时间
                    }
                    //
                    this.heartBeatCheckAndReallocateTaskTimer.schedule(
                            new PauseOrResumeTimerTask(this, this.heartBeatCheckAndReallocateTaskTimer,
                                    PauseOrResumeTimerTask.type_pause, planRunEndTimeString),//在这个时间点开始调用
                            endTimeAfterStartTimeAfterNowDate);
                    this.taskProcessorInfo.setNextRunEndTime(DateFormats.transferDataToString(endTimeAfterStartTimeAfterNowDate));
                    //
                } catch (Exception e)
                {
                    log.error("" + taskProcessorInfo.getTaskProcessorId(), e);
                    throw new Exception("" + taskProcessorInfo.getTaskProcessorId(), e);
                }
            }
        }
        if (isRunNow)
        {
            this.resume();
        }
        this.clearTaskItemListInTaskItemsProcessorAndRegisterTaskProcessorWithSerialNumberIfFailRefreshHeartBeatTimestampAndVersion();

    }

    /**
     * 超过运行的运行时间，暂时停止调度
     * <p>
     * cause:从生产环境实践发现，在zk不稳定且单机job多、调度间隔小的情况下。大量的线程不能正确
     * 地执行pause方法。官方3.3.3.2版本仍未处理此问题，此处个人处理办法是在暂停调度时加
     * 锁改为顺序执行。
     */
    //暂停操作:1时间表达式计算出应该暂停 2没有数据的时候暂停
    public void pause() throws Exception//pauseTaskProcessorAllocationStrategy!=stopScheduleFlag
    {
        if (this.isPauseSchedule == false)
        {
            this.isPauseSchedule = true;
            //

            if (this.taskDataItemProcessor != null)
            {
                this.taskDataItemProcessor.setStopScheduleFlagAndClearTaskList();
            }
            //暂停任务执行，同时还需要保证一个有个处理器存在.
            clearTaskItemListInTaskItemsProcessorAndRegisterTaskProcessorWithSerialNumberIfFailRefreshHeartBeatTimestampAndVersion();
        }
    }

    /**
     * 处在了可执行的时间区间，恢复运行
     * <p>
     * cause:同暂停方法，恢复方法也改为顺序执行
     */
    public void resume() throws Exception
    {
        if (this.isPauseSchedule == true)
        {

            this.isPauseSchedule = false;
            //
            if (this.taskDataItemCollectorAndExecuter != null)
            {
                if (this.taskUserDefinition.getProcessorType() != null && this.taskUserDefinition.getProcessorType().equalsIgnoreCase("ForEverRun") == true)
                {
                    this.taskUserDefinition.setProcessorType("ForEverRun");
                    this.taskDataItemProcessor = new ForEverRunTaskDataItemProcessor(this, taskDataItemCollectorAndExecuter, this.statistics);
                } else
                {
                    //WaitAllFinish
                    this.taskDataItemProcessor = new WaitAllFinishTaskDataItemProcessor(this, taskDataItemCollectorAndExecuter, this.statistics);
                    this.taskUserDefinition.setProcessorType("WaitAllFinish");
                }
            }
            clearTaskItemListInTaskItemsProcessorAndRegisterTaskProcessorWithSerialNumberIfFailRefreshHeartBeatTimestampAndVersion();
        }
    }

    /**
     * TaskProcessor状态监测。
     * 1开始init线程[末]
     * 2心跳线程[初][末]
     * 3计算第一次任务开始运行时间和结束时间[末]
     * 4终止任务[末]
     * 5恢复任务[末]
     */
    public void clearTaskItemListInTaskItemsProcessorAndRegisterTaskProcessorWithSerialNumberIfFailRefreshHeartBeatTimestampAndVersion() throws Exception
    {
        registerOrUnRegisterTaskProcessorLock.lock();
        try
        {
            if (this.isStopSchedule == true)
            {
                return;
            }
            if (!this.taskAndTaskProcessorAndTaskDataItemManage.setRegisterFalseIfNodeIsNotExistedOtherwiseUpdateHeartBeatInfo(this.taskProcessorInfo))
            {
                //更新信息失败，清除内存数据后重新注册
                this.clearTaskDataItemListInTaskDataItemsProcessor();
                this.taskAndTaskProcessorAndTaskDataItemManage.registerTaskProcessorInfoWithSerialNumberAndSetRegisterTrue(this.taskProcessorInfo);
            }
        } finally
        {
            registerOrUnRegisterTaskProcessorLock.unlock();
        }
    }

    public void clearTaskDataItemListInTaskDataItemsProcessor()
    {
        try
        {
            // 清除内存中所有的已经取得的数据和任务队列,在心态更新失败，或者发现注册中心的调度信息被删除
            this.taskItemStringAndParameterStringList.clear();
            if (this.taskDataItemProcessor != null)
            {
                //设置内存里面的任务数据需要重新装载
                this.taskDataItemProcessor.clearTaskDataItemList();
            }
        } finally
        {

            this.needGetTaskItemStringAndParameterStringList = true;
        }

    }

    public void unRegisterTaskProcessor() throws Exception
    {
        registerOrUnRegisterTaskProcessorLock.lock();
        try
        {
            if (this.taskDataItemProcessor != null)
            {
                this.taskDataItemProcessor = null;
            }
            //
            if (this.isPauseSchedule == true)// 是暂停调度，不注销Manager自己
            {
                return;
            }
            this.isStopSchedule = true;
            //
            // 取消心跳检测
            this.heartBeatCheckAndReallocateTaskTimer.cancel();
            // 从配置中心注销自己
            this.taskAndTaskProcessorAndTaskDataItemManage.deleteTaskProcessor(this.taskProcessorInfo.getTaskId(), this.taskProcessorInfo.getTaskProcessorId());
        } finally
        {
            registerOrUnRegisterTaskProcessorLock.unlock();
        }
    }
//

    /**
     * 定时向数据配置中心更新当前服务器的心跳信息。
     * 如果发现本次更新的时间如果已经超过了，服务器死亡的心跳周期，则不能在向服务器更新信息。
     * 而应该当作新的服务器，进行重新注册。
     */
    public void heartBeatCheckAndReallocateTaskItems() throws Exception
    {
        try
        {
            clearTaskItemListInTaskItemsProcessorAndRegisterTaskProcessorWithSerialNumberIfFailRefreshHeartBeatTimestampAndVersion();
            //如果任务信息没有初始化成功，不做任务相关的处理
            if (!this.minTaskProcessorIdIsControlTaskProcessorId)
            {
                return;
            }
            //重新分配任务(相当于主线程另起一个线程进行进行任务分配)
            this.reallocateTaskItemsTaskCurrentTaskProcessorIdIfCurrentTaskProcessorIdIsLeaderTaskProcessorId();
            //判断是否需要重新加载任务队列，避免任务处理进程不必要的检查和等待
            boolean needGetTaskItemStringAndParameterStringList = this.needGetTaskItemStringAndParameterStringList();
            //只要不相同，就设置需要重新装载，因为在心跳异常的时候，做了清理队列的事情，恢复后需要重新装载。
            if (needGetTaskItemStringAndParameterStringList != this.needGetTaskItemStringAndParameterStringList)
            {
                synchronized (getTaskItemStringAndParameterStringListLock)
                {
                    this.needGetTaskItemStringAndParameterStringList = true;
                }
                clearTaskItemListInTaskItemsProcessorAndRegisterTaskProcessorWithSerialNumberIfFailRefreshHeartBeatTimestampAndVersion();
            }
            //
            if (this.isPauseSchedule || (this.taskDataItemProcessor != null && taskDataItemProcessor.isSleeping()))
            {
                //如果当前线程组暂停了或者休眠中,需要释放该处理器,具体是将request_task_processor_id提升为current_task_processor_id
                this.synchronizedGetTaskItemStringAndParameterStringList();
            }
        } catch (Throwable e)
        {
            //清除内存中所有的已经取得的数据和任务队列,避免心跳线程失败时候导致的数据重复
            this.clearTaskDataItemListInTaskDataItemsProcessor();
            if (e instanceof Exception)
            {
                throw (Exception) e;
            } else
            {
                throw new Exception(e.getMessage(), e);
            }
        }
    }


    /**
     * 根据当前调度服务器的信息，重新计算分配所有的调度任务
     * 任务的分配是需要加锁，避免数据分配错误。为了避免数据锁带来的负面作用，通过版本号来达到锁的目的
     * <p>
     * 1、获取任务状态的版本号
     * 2、获取所有的服务器注册信息和任务队列信息
     * 3、清除已经超过心跳周期的服务器注册信息
     * 3、重新计算任务分配
     * 4、更新任务状态的版本号【乐观锁】
     * 5、根系任务队列的分配信息
     */

    public void reallocateTaskItemsTaskCurrentTaskProcessorIdIfCurrentTaskProcessorIdIsLeaderTaskProcessorId() throws Exception
    {
        taskAndTaskProcessorAndTaskDataItemManage.clearExpireTaskProcessor(this.taskProcessorInfo.getTaskId(), this.taskUserDefinition.getJudgeDeadInterval());
        //当前有效的taskProcessorIdList
        List<String> taskProcessorIdList = taskAndTaskProcessorAndTaskDataItemManage.getTaskProcessorIdList(this.taskProcessorInfo.getTaskId());

        if (!taskAndTaskProcessorAndTaskDataItemManage.isLeaderTaskProcessorId(taskProcessorIdList, this.taskProcessorInfo.getTaskProcessorId()))
        {
            return;
        }
        //
        //设置初始化成功标准，避免在leader转换的时候，新增的线程组初始化失败
        taskAndTaskProcessorAndTaskDataItemManage.setTaskItemsLeaderTaskProcessorId(this.taskProcessorInfo.getTaskId(), this.taskProcessorInfo.getTaskProcessorId());
        taskAndTaskProcessorAndTaskDataItemManage.clearTaskItemCurrentTaskProcessorIdNodeWhenNotFoundInTaskProcessorIdList(this.taskProcessorInfo.getTaskId(), taskProcessorIdList);
        taskAndTaskProcessorAndTaskDataItemManage.reallocateTaskItemsTaskCurrentTaskProcessorId(this.taskProcessorInfo.getTaskId(), taskProcessorIdList, this.taskProcessorInfo.getTaskProcessorId());
    }

    /**
     * 在leader重新分配任务，在每个server释放原来占有的任务项时，都会修改这个版本号
     */
    public boolean needGetTaskItemStringAndParameterStringList()
    {
        try
        {
            return this.lastFetchVersion < this.taskAndTaskProcessorAndTaskDataItemManage.getTaskProcessors$Stat$Version(this.taskProcessorInfo.getTaskId());
        } catch (Exception e)
        {
            throw new RuntimeException(e);
        }

    }

    /**
     * 该方法被任务项处理器调用
     * <br/>
     * 重新加载当前服务器的任务队列
     * 1、释放当前服务器持有，但有其它服务器进行申请的任务队列
     * 2、重新获取当前服务器的处理队列
     * <p>
     * 为了避免此操作的过度，阻塞真正的数据处理能力。系统设置一个重新装载的频率。例如1分钟
     * <p>
     * 特别注意：
     * 此方法的调用必须是在当前所有任务都处理完毕后才能调用，否则是否任务队列后可能数据被重复处理
     */
    public List<TaskItemStringAndParameterString> waiterunningTaskItemListIsAllFinshedAndSynchronizedGetTaskItemStringAndParameterStringList()
    {
        try
        {
            if (this.needGetTaskItemStringAndParameterStringList)
            {
                //特别注意：需要判断数据队列是否已经空了，否则可能在队列切换的时候导致数据重复处理
                //主要是在线程不休眠就加载数据的时候一定需要这个判断
                if (this.taskDataItemProcessor != null)
                {
                    while (!this.taskDataItemProcessor.runningTaskItemListIsAllFinshed())
                    {
                        Thread.sleep(50);
                    }
                }
                //真正开始处理数据
                synchronized (getTaskItemStringAndParameterStringListLock)
                {
                    this.synchronizedGetTaskItemStringAndParameterStringList();
                    this.needGetTaskItemStringAndParameterStringList = false;
                }
            }
            this.getTaskItemStringAndParameterStringListTimeMillis = this.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis();
            return this.taskItemStringAndParameterStringList;
        } catch (Exception e)
        {
            throw new RuntimeException(e);
        }
    }


    public boolean continueWhenHasData() throws Exception
    {
        if (!pauseWhenNotHasData())
        {
            return true;
        } else
        {
            this.pause();//没有数据,暂停调度
            return false;
        }
    }

    public boolean pauseWhenNotHasData()
    {

        //分配到任务(&&非立即执行)
        if (this.taskItemStringAndParameterStringList.size() > 0 && this.taskUserDefinition.getPlanRunStartTimeString() != null)
        {
            //可以立即终止
            if (this.taskUserDefinition.getPlanRunEndTimeString() == null
                    || this.taskUserDefinition.getPlanRunEndTimeString().equals("-1"))//代表需要运行完毕后终止
            {
                //有数据,运行完毕后,
                return true;//end
            } else
            {
                return false;
            }
        } else
        {
            return false;
        }
        //end return false
    }


}

